package net.bwie.douyin.dwd.log1.Function;


import net.bwie.douyin.dwd.log1.bean.AlertGiftSender;
import net.bwie.douyin.dwd.log1.bean.TikTokBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：realTimeGiftAlertFunction
 * @Author: ZddddQ
 * @Date: 2025/5/27 21:04
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class realTimeGiftAlertFunction implements WindowFunction<TikTokBean, AlertGiftSender, String, TimeWindow> {


    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow window, Iterable<TikTokBean> input, Collector<AlertGiftSender> out) throws Exception {

        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());

        // 初始化总金额
        double totalAmount = 0.0;

        // 遍历窗口内所有数据，累加金额
        for (TikTokBean bean : input) {
            totalAmount += bean.getGiftValue().doubleValue();
        }

        // 判断总金额是否超过阈值
        if (totalAmount > 10) {
            // 创建警报对象并填充数据
            AlertGiftSender alert = new AlertGiftSender();
            alert.setUserId(s);
            alert.setWindowStartTime(windowStartTime);
            alert.setWindowEndTime(windowEndTime);
            alert.setSendMoney(totalAmount);
            alert.setTs(System.currentTimeMillis());
            // 输出结果
            out.collect(alert);
        }
    }
}
